package com._119yn.host.gst200.config;

import com._119yn.host.gst200.mqtt.MqttChannels;
import com._119yn.host.gst200.mqtt.MqttMessageReceiver;
import com._119yn.host.gst200.mqtt.impl.MqttControlErrorMessageHandler;
import com._119yn.host.gst200.mqtt.impl.MqttControlMessageHandler;
import com._119yn.host.gst200.properties.MqttProperties;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.Mqttv5ClientManager;
import org.springframework.integration.mqtt.inbound.Mqttv5PahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.Mqttv5PahoMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.nio.charset.StandardCharsets;

@Slf4j
@Configuration
@EnableConfigurationProperties(MqttProperties.class)
@RequiredArgsConstructor
public class MqttConfig {

    /**
     * MQTT属性
     */
    private final MqttProperties properties;

    /**
     * MQTT连接选项
     *
     * @return MqttConnectOptions
     */
    @Bean
    public MqttConnectionOptions mqttConnectionOptions() {
        MqttConnectionOptions options = new MqttConnectionOptions();
        options.setServerURIs(properties.getBrokerUrls());
        if (properties.getCleanStart() != null) {
            options.setCleanStart(properties.getCleanStart());
        }
        if (properties.getConnectionTimeout() != null) {
            options.setConnectionTimeout(properties.getConnectionTimeout() * 1000);
        }
        if (properties.getKeepAliveInterval() != null) {
            options.setKeepAliveInterval(properties.getKeepAliveInterval());
        }
        if (properties.getAutomaticReconnect() != null) {
            options.setAutomaticReconnect(properties.getAutomaticReconnect());
        }
        if (properties.getUsername() != null && !properties.getUsername().isEmpty()) {
            options.setUserName(properties.getUsername());
        }
        if (properties.getPassword() != null && !properties.getPassword().isEmpty()) {
            options.setPassword(properties.getPassword().getBytes(StandardCharsets.UTF_8));
        }
        // 可以设置其他选项，如 SSL/TLS
        // options.setSocketFactory(...);
        return options;
    }

    @Bean
    public Mqttv5ClientManager mqttv5ClientManager(MqttConnectionOptions mqttConnectionOptions) {
        return new Mqttv5ClientManager(mqttConnectionOptions, properties.getClientId());
    }

    /**
     * 定义入站通道 (用于接收 MQTT 消息)
     *
     * @return MessageChannel
     */
    @Bean(MqttChannels.MQTT_HOST_CONTROL_MESSAGE_CHANNEL)
    public MessageChannel mqttHostControlMessageChannel() {
        return new DirectChannel();
    }

    /**
     * 定义入站错误通道 (用于接收 MQTT 消息)
     *
     * @return MessageChannel
     */
    @Bean(MqttChannels.MQTT_HOST_CONTROL_ERROR_MESSAGE_CHANNEL)
    public MessageChannel mqttHostControlErrorMessageChannel() {
        return new DirectChannel();
    }

    /**
     * MQTT 消息驱动的入站通道适配器 (订阅主题)
     *
     * @return MessageProducer
     */
    @Bean
    public MessageProducer mqttHostControlMessageProducer(
            Mqttv5ClientManager mqttv5ClientManager,
            MessageChannel mqttHostControlMessageChannel,
            MessageChannel mqttHostControlErrorMessageChannel
    ) {
        Mqttv5PahoMessageDrivenChannelAdapter mqttv5PahoMessageDrivenChannelAdapter
                = new Mqttv5PahoMessageDrivenChannelAdapter(mqttv5ClientManager, properties.getDefaultTopic());
        mqttv5PahoMessageDrivenChannelAdapter.setOutputChannel(mqttHostControlMessageChannel);
        mqttv5PahoMessageDrivenChannelAdapter.setErrorChannel(mqttHostControlErrorMessageChannel);
        for (MqttProperties.Topic topic : properties.getTopics().getSubscribe().getHostControl()) {
            mqttv5PahoMessageDrivenChannelAdapter.addTopic(
                    topic.getName(),
                    topic.getQos() != null ? topic.getQos() : properties.getDefaultQos()
            );
        }

        return mqttv5PahoMessageDrivenChannelAdapter;
    }

    /**
     * 定义出站通道 (用于发送 MQTT 消息)
     *
     * @return MessageChannel
     */
    @Bean(MqttChannels.MQTT_HOST_CONTROL_ACK_MESSAGE_CHANNEL)
    public MessageChannel mqttHostControlAckMessageChannel() {
        return new DirectChannel();
    }

    /**
     * MQTT 消息处理程序 (用于发布消息)
     *
     * @return MessageHandler
     */
    @Bean
    @ServiceActivator(inputChannel = MqttChannels.MQTT_HOST_CONTROL_ACK_MESSAGE_CHANNEL)
    public MessageHandler mqttHostControlAckMessageProducer(
            Mqttv5ClientManager mqttv5ClientManager
    ) {
        Mqttv5PahoMessageHandler mqttv5PahoMessageHandler = new Mqttv5PahoMessageHandler(mqttv5ClientManager);
        mqttv5PahoMessageHandler.setDefaultTopic(properties.getDefaultTopic());
        mqttv5PahoMessageHandler.setDefaultQos(properties.getDefaultQos());
        return mqttv5PahoMessageHandler;
    }

    /**
     * 定义出站通道 (用于信号上传)
     *
     * @return MessageChannel
     */
    @Bean(MqttChannels.MQTT_HOST_SIGNAL_MESSAGE_CHANNEL)
    public MessageChannel mqttHostSignalMessageChannel() {
        return new DirectChannel();
    }

    /**
     * MQTT 消息处理程序 (用于发布消息)
     *
     * @return MessageHandler
     */
    @Bean
    @ServiceActivator(inputChannel = MqttChannels.MQTT_HOST_SIGNAL_MESSAGE_CHANNEL)
    public MessageHandler mqttHostSignalMessageHandler(Mqttv5ClientManager mqttv5ClientManager) {
        Mqttv5PahoMessageHandler mqttv5PahoMessageHandler = new Mqttv5PahoMessageHandler(mqttv5ClientManager);
        mqttv5PahoMessageHandler.setDefaultTopic(properties.getDefaultTopic());
        mqttv5PahoMessageHandler.setDefaultQos(properties.getDefaultQos());
        return mqttv5PahoMessageHandler;
    }

    @Bean
    @ServiceActivator(inputChannel = MqttChannels.MQTT_HOST_CONTROL_MESSAGE_CHANNEL)
    public MessageHandler mqttHostControlMessageHandler(MqttMessageReceiver mqttMessageReceiver) {
        return new MqttControlMessageHandler(mqttMessageReceiver);
    }

    @Bean
    @ServiceActivator(inputChannel = MqttChannels.MQTT_HOST_CONTROL_ERROR_MESSAGE_CHANNEL)
    public MessageHandler mqttHostControlErrorMessageHandler(MqttMessageReceiver mqttMessageReceiver) {
        return new MqttControlErrorMessageHandler(mqttMessageReceiver);
    }

}
